package com.aotain.metis.bolt;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;


/**
 * 
 * SQL注入检测
 * 发现源IP发起的SQL注入行为 [A->B的攻击行为]
 * 
 * @author  Turk
 * @version  [版本号, 2017年2月14日]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class SQLInjectionBoltMetis extends AbstractBolt{


	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> _mapStreamoctets = new HashMap<String,Long>();
	
	/**
	 * 上行包数
	 */
	private HashMap<String,Long> _mapStreampacket = new HashMap<String,Long>();
	
	private static final long serialVersionUID = 1L;
	private Map<String,String> tupleMap = new HashMap<String,String>();
	private Map<String,Integer> sqlNumMap = new HashMap<String,Integer>();
	private int threshold = 3;
	public String injectionKeyList = null;

	public SQLInjectionBoltMetis(){
		this.threshold = 5;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d","dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour", "abntype", "idcno", "sourcecountry","streamoctets","streampacket"));
		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport","accesstime","sourcearea",
				"sourcegeo","sourcecountry","destarea","destgeo","evaluate","abnormal","desc","attnum","delaytime","flag"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {		
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}
	
	 
	private int  matchRuleBase(String url){
		
		int matchNum = 0;
		try{				
			if(url.isEmpty()) return matchNum;
			//url = URLDecoder.decode(CommonFunction.decodeBASE64(url),"utf-8").toLowerCase();
			url = url.toLowerCase();
			if(url.indexOf("?") == -1) return matchNum;
			url = url.substring(url.indexOf("?"),url.length()).replaceAll("'", "");
			String rgx = "\\b(" + injectionKeyList + ")\\b";
			Pattern pattern = Pattern.compile(rgx); 
			Matcher m = pattern.matcher(url);
			ArrayList<String> list = new ArrayList<String>(); 
			while(m.find()) {
				if(!list.contains(m.group())) list.add(m.group());
			}
			
			if(list.contains("order") && !list.contains("by")){
				matchNum = 1;
			}else if(list.contains("group") && !list.contains("by")){
				matchNum = 1;
			}else{
				matchNum =  list.size();
			}
		}catch(Exception ex){
			Logger.getRootLogger().error("=====SQLInjectionBolt:matchRuleBase=========ERROR=================",ex);
		}		
		return matchNum;
	
	}

	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try 
		{
			String url = tuple.getStringByField("url").trim();		
			//Logger.getRootLogger().info("tuple======"+tuple);
			if(matchRuleBase(url) >= 2){
				
				String type =  tuple.getStringByField("type").trim();	
				String dip = tuple.getStringByField("dip");
				String dport = tuple.getStringByField("dport");
				String sip = tuple.getStringByField("sip");			
				String sport = tuple.getStringByField("sport");
				String gis = new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
				String idc = tuple.getStringByField("idcno");		
				String Key = dip + "|" + dport + "|" + sip + "|" + type ;
				//url = CommonFunction.decodeBASE64(url).toLowerCase();				
				url = url.toLowerCase();
				if(sqlNumMap.containsKey(Key)) {				
					sqlNumMap.put(Key, sqlNumMap.get(Key) + 1);
				}else{
					sqlNumMap.put(Key,1);
				}
				tupleMap.put(Key, dip + "|" + dport + "|" + sip + "|" + sport +"|" + gis +  "|"  + type + "|"  + idc);
				
				if(tuple.contains("upstreamoctets") && tuple.contains("upstreampacket")) {
					long nUpStreamOctets = tuple.getLongByField("upstreamoctets");
					long nUpstreampacket = tuple.getLongByField("upstreampacket");
					if(_mapStreamoctets.containsKey(Key)) {
						nUpStreamOctets = _mapStreamoctets.get(Key) + nUpStreamOctets;
					}
					_mapStreamoctets.put(Key, nUpStreamOctets);
					
					if(_mapStreampacket.containsKey(Key)){
						nUpstreampacket = _mapStreampacket.get(Key) + nUpstreampacket;
					}
					_mapStreampacket.put(Key, nUpstreampacket);
				
				}
			}
			
		} catch(Exception ex) {
			Logger.getLogger(SQLInjectionBoltMetis.class).error("=====countInLocal=========ERROR=================",ex);
		}
	}
	


	//定时发送
	private void emitCountingData(OutputCollector collector) {
		
		try{
			HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
			String info = String.format("######DEBUG:SQLInjection MAP[%d]",sqlNumMap.size());
			Logger.getLogger(SQLInjectionBoltMetis.class).info(info);
			
			for(String dkey : sqlNumMap.keySet()) {	
				long urlPVNum = sqlNumMap.get(dkey);
				if(urlPVNum >= threshold){
					System.out.println("SQLInjection target >>>>>>>>> "+dkey + " >>>>>>>>> matchNum:" +  sqlNumMap.get(dkey));
					
					if(tupleMap.containsKey(dkey)) {
						String[] tuplesp = tupleMap.get(dkey).split("\\|",-1);
						String dip = tuplesp[0];
						String dport = tuplesp[1];
						String sip = tuplesp[2];
						String sport = tuplesp[3];
						String gis = tuplesp[4];
						String type = tuplesp[5];
						String idc = tuplesp[6];
						String[] gisArray = gis.split("#");
						String destAreaName = gisArray[0];
						String destGis = gisArray[1];
						String sourceAreaName = gisArray[2];
						String sourceGis = gisArray[3];
						String sourceAreaCountry = gisArray[4];
						String sourceAreaId = gisArray[5];
						String sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
//						String sourceAreaCityId = gisArray[7];
//						String sourceAreaProvinceId = gisArray[8];
				
						String sAbnormalTable = "METIS_ABNORMAL_LOG";
						
						String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()+15000));
						//源IP发起的攻击行为
						String rowKeyAbnormal = sip + "_"  + accesstime + "_" + dip + "_" + dport + "_8";	
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "SOURCEIP", sip);
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "DESTPORT", dport);							
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "ABRNORMAL", "8");
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "DESTIP", dip);							
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "DESTGEO", destGis);							
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(urlPVNum));
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "DESC", "SQL Injection");
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "EVALUATE", "40");	
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "ENDTIME", accesstime);
						if(sourceProvinceName.equals("")) sourceProvinceName = sourceAreaCountry;
						hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
						
				    	//UDII数据特有
						long nUpStreamOctets = 0;
						long nUpstreampacket = 0;
						if(_mapStreamoctets.size() > 0 && _mapStreampacket.size() > 0)
						{
							nUpStreamOctets = _mapStreamoctets.get(dkey);
							nUpstreampacket = _mapStreampacket.get(dkey);
							hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
							hbaseInstance.Add(sAbnormalTable, rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
							urlPVNum = nUpStreamOctets;
						}
						
						Date current = new Date();
						SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
						SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
						SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
				    	String dateStr_h = sdf_h.format(current);
						String dateStr_d = sdf_d.format(current);
						String dateStr_hour = sdf_hour.format(current);
						
						collector.emit(new Values(dip,sip,dport,sport,dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
				    			sourceProvinceName, urlPVNum, dateStr_hour,"SQLINJECTION" , idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
				    	
						
					}
				}
			}
		}catch(Exception e){
			
		}finally{
			sqlNumMap.clear();
			tupleMap.clear();
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(tuple)) {
				emitCountingData(collector);  
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("SQLInjectionBolt throws an exception!");
		}	
	}
	

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
		String sqlInjection = stormConf.get("sqlinjection").toString();	
		Logger.getLogger(SQLInjectionBoltMetis.class).info("sqlInjection>>>"+sqlInjection);
		
		injectionKeyList = sqlInjection.replace("-", "=");	
		Logger.getLogger(SQLInjectionBoltMetis.class).info(">>>>injectionKeyList>>>>" +injectionKeyList);
	}

}
